package org.sxp.common.utils.datax.writer;

import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Maps;
import org.sxp.common.dto.datax.reader.DataXRDMSReader;
import org.sxp.common.dto.datax.writer.DataXRDMSWriter;
import org.sxp.common.enums.DbTypeEnum;
import org.sxp.common.utils.SpringContextUtils;
import org.sxp.common.validator.ValidatorUtils;
import org.sxp.common.validator.group.HbaseWriterGroup;
import org.sxp.modules.datasource.entity.DatabaseDO;
import org.sxp.modules.datasource.service.DatabaseService;
import org.sxp.modules.dataworks.dto.SyncDTO;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * datax writer base
 *
 * @author zhouhongfa@gz-yibo.com
 * @ClassName BaseWriterPlugin
 * @Version 1.0
 * @since 2019/8/2 16:28
 */
public abstract class BaseWriter implements IDataxWriter {
    private static final Pattern p = Pattern.compile("\r\n|\r|\n|\n\r");

    protected final Pattern pattern = Pattern.compile("\"writer\":(\\{.*)}],");

    protected SyncDTO syncDTO;

    public BaseWriter(SyncDTO syncDTO){
        this.syncDTO = syncDTO;
    }

    protected void beforeBuild() {
        if(syncDTO.getTargetDbType().equals(DbTypeEnum.hbase.getCode())){
            ValidatorUtils.validateEntity(syncDTO, HbaseWriterGroup.class);
        }
    }

    @Override
    public Map<String, Object> build() {
        DatabaseService databaseService = SpringContextUtils.getBean(DatabaseService.class);
        DatabaseDO jobDatasource = databaseService.getById(syncDTO.getTargetDb());
        Map<String, Object> writerObj = Maps.newLinkedHashMap();
        writerObj.put("name", getName());
        List<String> columnList = new ArrayList<>();
        for (JSONObject columnObj : syncDTO.getTargetField()) {
            columnList.add(columnObj.getString("name"));
        }
        Map<String, Object> parameterObj = Maps.newLinkedHashMap();

        parameterObj.put("username", jobDatasource.getUsername());
        parameterObj.put("password", jobDatasource.getPassword());
        parameterObj.put("column", columnList);
        parameterObj.put("preSql", splitSql(syncDTO.getPreSql()));
        parameterObj.put("postSql", splitSql(syncDTO.getPostSql()));
        if(StrUtil.isNotBlank(syncDTO.getWriteMode())){
            parameterObj.put("writeMode", syncDTO.getWriteMode());
        }

        Map<String, Object> connectionObj = Maps.newLinkedHashMap();
        connectionObj.put("table", new String[]{syncDTO.getTargetTable()});
        connectionObj.put("jdbcUrl", jobDatasource.getUrl());

        parameterObj.put("connection", ImmutableList.of(connectionObj));
        writerObj.put("parameter", parameterObj);

        return writerObj;
    }

    @Override
    public void backBuild() {
        String removeSpace = StrUtil.trim(syncDTO.getDataxJson());
        /*
         * 根据key和josn字符串取出特定的value :{"(.*?)"}
         */
        Matcher matcher = pattern.matcher(removeSpace);
        String value = null;
        if (matcher.find()) {
            value = matcher.group(1);
        }
        if(JSONUtil.isJson(value)){
            DataXRDMSWriter writer = JSON.parseObject(value, DataXRDMSWriter.class);
            if(ArrayUtil.isNotEmpty(writer.getParameter().getPostSql())){
                syncDTO.setPostSql(String.join(";", writer.getParameter().getPostSql()));
            }
            if(ArrayUtil.isNotEmpty(writer.getParameter().getPreSql())){
                syncDTO.setPreSql(String.join(";", writer.getParameter().getPreSql()));
            }
        }

        afterBackBuild();
    }

    protected void afterBackBuild(){

    }

    private String[] splitSql(String sql) {
        String[] sqlArr = null;
        if (StrUtil.isNotBlank(sql)) {
            Matcher m = p.matcher(sql);
            String sqlStr = m.replaceAll("");
            sqlArr = sqlStr.split(";");
        }
        return sqlArr;
    }
}
